/**
 *    Copyright (c) [2022] [中和农信]
 *    rabbit-mq-plus is licensed under Mulan PSL v2.
 *    You can use this software according to the terms and conditions of the Mulan PSL v2.
 *    You may obtain a copy of Mulan PSL v2 at:
 *             http://license.coscl.org.cn/MulanPSL2
 *    THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 *    See the Mulan PSL v2 for more details.
 *
 */
package com.cfpamf.athean.rabbit.mq.plus.consumer.core.impl;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.Resource;

import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import com.cfpamf.athean.rabbit.mq.plus.common.constant.BusinessConstants;
import com.cfpamf.athean.rabbit.mq.plus.common.enums.ProcessStatusEnum;
import com.cfpamf.athean.rabbit.mq.plus.consumer.core.schedule.RabbitMqPlusConsumerFailureRetryJob;
import com.cfpamf.athean.rabbit.mq.plus.consumer.listener.RabbitMqPlusConsumerListener;
import com.cfpamf.athean.rabbit.mq.plus.domain.dto.RabbitMqPlusConsumerDTO;
import com.cfpamf.athean.rabbit.mq.plus.domain.dto.RabbitMqPlusConsumerDetailDTO;
import com.cfpamf.athean.rabbit.mq.plus.domain.dto.RabbitMqPlusConsumerTaskDTO;
import com.cfpamf.athean.rabbit.mq.plus.service.RabbitMqPlusConsumerTaskService;
import com.cfpamf.athena.utils.DateUtils;
import com.cfpamf.athena.utils.JSONUtils;
import com.cfpamf.athena.utils.ObjectUtils;
import com.cfpamf.athena.utils.StringUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;

/**
 * 类DefaultRabbitMqPlusConsumerFailureRetryJob.java的实现描述：
 *
 * @author Andy（陈洪桥）2022年12月12 21:47
 */
@Slf4j
@Component("rabbitMqPlusConsumerFailureRetryJob")
public class DefaultRabbitMqPlusConsumerFailureRetryJob implements RabbitMqPlusConsumerFailureRetryJob {

    volatile Lock lock = new ReentrantLock();

    @Resource
    private RabbitMqPlusConsumerTaskService rabbitMqPlusConsumerTaskService;

    @Resource
    private RabbitMqPlusConsumerHandler rabbitMqPlusConsumerHandler;

    @Lazy
    @Resource
    private Map<String, RabbitMqPlusConsumerListener> consumerListenerMap = Maps.newConcurrentMap();

    @Override
    public void runAutoRetryCheck() {
        try {
            if (!lock.tryLock(BusinessConstants.FIVE, TimeUnit.SECONDS)) {// 重入锁，等待5s
                if (log.isInfoEnabled()) {
                    log.info("[rabbitMqPlus]启用rabbitMq消费者消息消费状态检查，执行时间：{} 上一批job还在处理中，拒绝执行。",
                        DateUtils.format(DateUtils.nowTime()));
                }
                return;
            }
            if (log.isInfoEnabled()) {
                log.info("[rabbitMqPlus]启用rabbitMq消费者消息消费状态检查，执行时间：{}", DateUtils.format(DateUtils.nowTime()));
            }
            List<Integer> status = Lists.newArrayList();
            status.add(ProcessStatusEnum.TODO.getCode());
            // status.add(ProcessStatusEnum.PROCESSING.getCode());
            List<RabbitMqPlusConsumerTaskDTO> data = rabbitMqPlusConsumerTaskService.queryTodoDataByStatus(status);
            for (RabbitMqPlusConsumerTaskDTO consumerTaskDTO : data) {
                // 检查TODO消费消息状态不等于完成，&&retryCount < maxRetryCount
                if (ProcessStatusEnum.isNotDone(consumerTaskDTO.getStatusCode())
                    && consumerTaskDTO.getRetryCount() <= consumerTaskDTO.getMaxRetryCount()) {
                    RabbitMqPlusConsumerDetailDTO body =
                        JSONUtils.jsonToBean(consumerTaskDTO.getMessageBody(), RabbitMqPlusConsumerDetailDTO.class);
                    RabbitMqPlusConsumerDTO consumerDTO =
                        ObjectUtils.copyPropertys(body, RabbitMqPlusConsumerDTO.class);
                    String className = consumerTaskDTO.getConsumerListenerImplClassName();
                    String classNameFirstCharacter = className.substring(0, 1);
                    // 首字母替换为小写
                    String instanceName = className.replaceFirst(classNameFirstCharacter,
                        StringUtils.toLowerCase(classNameFirstCharacter));
                    consumerDTO.setRabbitMqPlusConsumerListener(consumerListenerMap.get(instanceName));
                    // 消息消费失败，补偿重新消费消息，更新DB重试次数和状态
                    rabbitMqPlusConsumerHandler.consumerProcessor(rabbitMqPlusConsumerTaskService, consumerDTO,
                        body.getConsumerTag(), (consumerTaskDTO.getRetryCount() + 1),
                        consumerTaskDTO.getFirstConsumerTime(), body.getMessageId(), body.getMessageBody());
                }
            }
        } catch (Exception e) {
            if (log.isInfoEnabled()) {
                log.error("[rabbitMqPlus]启用rabbitMq消费者消息消费状态检查，错误信息异常：{} ", e);
            }
        } finally {
            lock.unlock();
        }
    }

}
